package com.atguigu.app

import java.text.SimpleDateFormat
import java.util.Date

import com.alibaba.fastjson.JSON
import com.atguigu.bean.StartUpLog
import com.atguigu.constants.GmallConstants
import com.atguigu.handle.DauHandle
import com.atguigu.utils.MyKafkaUtil
import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.phoenix.spark._
/**
  *
  * @author Lec
  * @date 2022/7/15 22:33
  */
object DauApp {
  def main(args: Array[String]): Unit = {
    //1.创建sparkConf
    val sparkConf: SparkConf = new SparkConf().setAppName("DauApp").setMaster("local[*]")
    //2.创建streamingContext
    val ssc = new StreamingContext(sparkConf,Seconds(3))

    //3.消费kafka数据
    val kafkaDStream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_STARTUP,ssc)
//    测试代码
//    kafkaDStream.foreachRDD(rdd =>{
//      rdd.foreachPartition(partititon =>{
//        partititon.foreach(record => {
//          println(record.value())
//        })
//      })
//    })


//    4.将从kafka过来的数据转为样例类，补全logDate和logHour字段
    val sdf = new SimpleDateFormat("yyyy-MM-dd HH")
    val startUpLogDStream: DStream[StartUpLog] = kafkaDStream.mapPartitions(partition => {
      partition.map(record => {
        //4.1将json字符串转为样例类
        val startUpLog: StartUpLog = JSON.parseObject(record.value(), classOf[StartUpLog])

        //4.2补全logDate&logHour字段
        //对样例类中的ts做格式化操作  yyyy-MM-dd HH
        val times: String = sdf.format(new Date(startUpLog.ts))
        startUpLog.logDate = times.split(" ")(0)
        startUpLog.logHour = times.split(" ")(1)
        startUpLog
      })
    })

    //只有对流（DStream）打印的时候会在控制台打印时间戳
//    startUpLogDStream.print()

    //一条流用到多次可以cache一下
    startUpLogDStream.cache()

    //打印原始数据条数
    startUpLogDStream.count().print()

    //5.批次间去重
    val filterByRedisDStream: DStream[StartUpLog] = DauHandle.filterBYRedis(startUpLogDStream,ssc.sparkContext)

    //一条流用到多次可以cache缓存一下
    filterByRedisDStream.cache()

    //打印经过批次间去重后的数据条数
    filterByRedisDStream.count().print()

    //6.批次内去重
    val filterByGroupDStream: DStream[StartUpLog] = DauHandle.filterByGroup(filterByRedisDStream)

    //打印经过批次内去重后的数据条数
    filterByGroupDStream.count().print()

    //7.将最终去重后的mid写入Redis，为了能够做批次间去重，通过对比Redis中mid来判断有没有重复
    DauHandle.saveMidToRedis(filterByGroupDStream)

    //8.将经过批次内去重后的明细数据写入HBASE（Phoenix）
    filterByGroupDStream.foreachRDD(rdd => {
      rdd.saveToPhoenix(
        "GMALL_DAU",
        Seq("MID", "UID", "APPID", "AREA", "OS", "CH", "TYPE", "VS", "LOGDATE", "LOGHOUR", "TS"),
        HBaseConfiguration.create,
        Some("hadoop102,hadoop103,hadoop104:2181")
      )
    })



    ssc.start()
    ssc.awaitTermination()
  }

}
